package SparkGraphXInAction

import org.apache.spark._
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.graphx._
import org.apache.spark.graphx.Graph._
import org.apache.spark.rdd.RDD
import org.apache.spark.graphx.util.GraphGenerators

/**
  * Created by Administrator on 2017/4/23 0023.
  * 算法的最终结果，我们希望为每个顶点标记上离它最远的根顶点的距离。
  * 假设图不存在环（环表示当沿着边从一个顶点出发，最终又回到同一个顶点）。
  * 处理包含环的图通常会增加算法的复杂性，我们稍后会介绍一些处理环的策略。
  * 首先我们对aggregateMessages会调用到的sendMsg和mergeMsg进行定义。我们不把sendMsg和mergeMsg作为匿名函数传给aggregateMessages函数，
  * 而是显式地定义sendMsg和mergeMsg函数，这样会使代码更为清晰。
  * 在函数式编程中实现迭代通常采用递归的方式，所以接下来我们会定义一个用于递归的辅助函数propagateEdgeCount，它会持续调用aggregateMessages。
  * propagate : 扩散，繁殖。
  */
object DistanceFromRootVertex {

  //sendMsg函数作为参数传入aggregateMessages。
  //这个函数会在图中的每条边上被调用。这里sendMsg只是简单的累加计数器。
  def sendMsg(ec:EdgeContext[Int,String,Int]): Unit = {
    ec.sendToDst(ec.srcAttr+1)
  }

  //这里我们定义了mergeMsg函数，这个函数会在所有的消息传递到顶点后被重复调用。
  //消息经过合并后，最终得出结果为包含最大距离值的顶点。
  def mergeMsg(a: Int, b: Int): Int = {
    math.max(a,b)
  }

  def propagateEdgeCount(g:Graph[Int, String]):Graph[Int, String] ={
    //生成新的顶点集
    val verts = g.aggregateMessages[Int](sendMsg, mergeMsg)
    //生成一个更新后的包含新的信息的图。
    val g2 = Graph(verts, g.edges)
    //让我们将两组顶点连接在一起来看看更新的图
    //是否有任何新的信息——这会生成新的数据
    //Tuple2[vertexId, Tuple2[old vertex data, new vertex data]]
    val check = g2.vertices.join(g.vertices).
      //查看join顶点集后的每个元素，并计算元素中的不同点，如果相同则返回0。
      map(x => x._2._1 - x._2._2).
      //合计所有的不同，如果所有的顶点完全相同，合计结果为0。
      reduce(_ + _)
    if (check > 0 )
      //如果有变化，则继续递归执行。
      propagateEdgeCount(g2)
    else
      //没有变化则返回传入的图对象。
      g
  }
  //在上述代码中，propagateEdgeCount()会将当前距离加1，并将它发送给每条边的目标定顶点。然后目标顶点对它接收到的所有消息进行max()操作，并进行
  //距离的更新。
  //值得注意的一个重点是，每轮迭代后，我们对原始图和更新图进行比较，来定义结束递归的条件。当这两个图没有差异时则结束迭代。这里起关键作用
  //的是reduce(_+_),因为我们知道更新后的顶点间的距离至少和原来的距离值一样大，因此它们的差不可能是负数。不会出现负数与正数相加最终得到0的情况。

  //现在我们把递归函数准备好了，将myGraph图作为输入，首先要进行初始化。这是任何迭代算法都需面对的关键问题：怎么开始呢？我们需要理清任务的需求，
  //开始的时候每个顶点已知的消息是什么，我们期待得到什么样的结果。我们期待的是一个在图中经过了最远距离的正数。在开始的时候，我们不知道距离值，所
  //以将每个顶点的值设置为0，让算法逐步在图中传播消息

  def main(args: Array[String]): Unit = {
    //设置运行环境
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    val myVertices = sc.makeRDD(Array((1L, "Ann"),(2L, "Bill"), (3L, "Charles"), (4L, "Diane"), (5L, "Went to gym this morning")))
    val myEdges = sc.makeRDD(Array(Edge(1L, 2L, "is-friends-with"), Edge(2L, 3L, "is-friends-with"),
      Edge(3L, 4L, "is-friends-with"), Edge(4L, 5L, "Likes-status"), Edge(3L, 5L, "Wrote-status")))
    val myGraph = Graph(myVertices, myEdges)

    val initialGraph = myGraph.mapVertices((_,_) => 0)
    val lll = propagateEdgeCount(initialGraph).vertices.collect
    for(vertice <- lll){
      println(vertice)
    }
    /*结果：
    (4,3)
    (1,0)
    (3,2)
    (5,4)
    (2,1)
     */
    //可以看到，实现一个迭代算法很方便。关键点在于要考虑信息（“消息”）是如何通过边来传播和累计，一边多次迭代后得到我们想要的结果的。
  }
}
